package com.yjy.RxJava;

import org.junit.Test;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func0;

import java.util.Arrays;
import java.util.List;

public class MainTest {

    /**
     * create就是创建一个Observable，来生产消息
     *
     * Observable 使用了onNext方法生产了2个水果，Apple和orange，然后调用了onCompleted方法结束了这次生产
     * 消费者用onnext方法收到了2个水果，所以消费者就将收到的水果打印出来了，没有做任何处理
     */
    @Test
    public void test1_create() {
        Observable<String> producer = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("blue");
                subscriber.onNext("red");
                subscriber.onNext("green");
                subscriber.onNext("yellow");
                subscriber.onNext("orange");
                subscriber.onNext("cyan");
                subscriber.onNext("purple");
                subscriber.onCompleted();
            }
        });
        Subscriber<String> consumer = new Subscriber<String>() {
            @Override
            public void onNext(String s) {
                System.out.println("我收到的颜色有 = " + s);
            }

            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
            }
        };
        producer.subscribe(consumer);
    }

    // 使用from简化上面操作
    @Test
    public void test2_from() {
        List<String> list = Arrays.asList("blue", "red", "green", "yellow", "orange", "cyan", "purple");
        Observable.from(list).subscribe(new Action1<String>() {
            @Override
            public void call(String color) {
                System.out.println("我收到的颜色有 = " + color);
            }
        });
    }

    // 使用from和lambda表达式简化上面操作
    @Test
    public void test3_from_lambda() {
        List<String> list = Arrays.asList("blue", "red", "green", "yellow", "orange", "cyan", "purple");
        Observable.from(list).subscribe(color -> System.out.println("我收到的颜色有 = " + color));
    }

    @Test
    public void test4_defer() {
        // 只有当订阅者订阅才创建Observable，为每个订阅创建一个新的Observable。内部通过OnSubscribeDefer在订阅时调用Func0创建Observable
        List<String> list = Arrays.asList("blue", "red", "green", "yellow", "orange", "cyan", "purple");
        Observable.defer(new Func0<Observable<String>>() {
            @Override
            public Observable<String> call() {
                return Observable.from(list);
            }
        }).subscribe(new Action1<String>() {
            @Override
            public void call(String color) {
                System.out.println("我收到的颜色有 = " + color);
            }
        });
    }

    // reduce是一个聚合函数，接收上次运算的结果
    @Test
    public void test_reduce() {
        List<Integer> list = Arrays.asList(10, 5, 3, 2, 1, 0);
        Observable.from(list).reduce((result, num) -> {
            System.out.println("开始前 result="+result+" , num=" + num);
            result+=num;
            return result;
        }).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer result) {
                System.out.println("开始前 result="+result);
            }
        });
    }

    // 测试滑动窗口
    @Test
    public void test_window() {
        System.out.println(">>>>>>>>>>>测试滑动窗口：每次滑动1个单位，3个单位为一整体");
        List<Integer> list = Arrays.asList(10, 5, 3, 2, 1, 0);
        Observable.from(list).window(2, 1).subscribe(new Action1<Observable<Integer>>() {
            @Override
            public void call(Observable<Integer> integerObservable) {
                integerObservable.reduce((sum, num) -> sum+=num).subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("我被2个打印一次 = " + integer);
                    }
                });
            }
        });

        System.out.println(">>>>>>>>>>>测试滑动窗口：每次滑动2个单位，3个单位为一整体");
        Observable.from(list).window(2, 2).subscribe(new Action1<Observable<Integer>>() {
            @Override
            public void call(Observable<Integer> integerObservable) {
                integerObservable.reduce((sum, num) -> sum+=num).subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println("我被2个打印一次 = " + integer);
                    }
                });
            }
        });

    }
}
